package com.dy.realtime.dwd;

import com.dy.realtime.common.base.BaseSqlApp;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dwd
 * @className: DwdDoctorRegister
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/12 11:20
 * @version: 1.0
 * 医生域医生注册事务事实表
 */

public class DwdDoctorRegister extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdDoctorRegister().start(Constant.TOPIC_DWD_TRADE_DOCTOR_REGISTER,Constant.TOPIC_DWD_TRADE_DOCTOR_REGISTER,4,10016);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
         readOdsDb(tableEnv,groupId);
        Table doctor = tableEnv.sqlQuery("select `data`['id'] id, \n" +
                "       `data`['create_time'] register_time,\n" +
                "       `data`['birthday'] birthday,\n" +
                "       `data`['consultation_fee'] consultation_fee,\n" +
                "       `data`['gender'] gender_code,\n" +
                "       `data`['specialty'] specialty_code,\n" +
                "       `data`['title'] title_code,\n" +
                "       `data`['hospital_id'] hospital_id\n" +
                "from topic_db_medical\n" +
                "where `table` = 'doctor'\n" +
                "  and `type` = 'insert';");

        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_DOCTOR_REGISTER+" (\n" +
                 "    id string,\n" +
                 "    register_time string,\n" +
                 "    birthday string,\n" +
                 "    consultation_fee string,\n" +
                 "    gender_code string,\n" +
                 "    specialty_code string,\n" +
                 "    title_code string,\n" +
                 "    hospital_id string\n" +
                 ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_DOCTOR_REGISTER));
         doctor.executeInsert(Constant.TOPIC_DWD_TRADE_DOCTOR_REGISTER);
    }
}
